package com.tanfuqi.sicent.server.filter

import com.tanfuqi.sicent.protocol.message.Message
import org.apache.mina.core.filterchain.IoFilter
import org.apache.mina.core.filterchain.IoFilterAdapter
import org.apache.mina.core.session.IoSession
import reactor.core.Disposable
import reactor.core.publisher.Flux
import reactor.core.publisher.FluxSink
import reactor.core.publisher.Mono
import reactor.core.publisher.toMono
import java.time.Duration
import java.util.concurrent.TimeoutException
import java.util.concurrent.atomic.AtomicInteger

open class ReceiveFilter : IoFilterAdapter() {
    override fun messageReceived(nextFilter: IoFilter.NextFilter?, session: IoSession?, message: Any?) {
        sink?.next(message as Message)
        super.messageReceived(nextFilter, session, message)
    }

//    override fun exceptionCaught(nextFilter: IoFilter.NextFilter?, session: IoSession?, cause: Throwable?) {
//
//        super.exceptionCaught(nextFilter, session, cause)
//    }
    companion object {
        private var sink: FluxSink<Message>? = null

        val receiveFlux: Flux<Message> = Flux.create {
            sink = it
        }
    }
}

val messageId = AtomicInteger(1)
fun Mono<IoSession>.sent(message: Message): Mono<Message> = message.toMono().flatMap { message ->
    this.flatMap { session ->
        message.id = messageId.getAndIncrement()
        val success = session.write(message).await(10000)
        if (success) {
            var disposable: Disposable? = null
            Mono.create<Message> { sink ->
                disposable = ReceiveFilter.receiveFlux.filter {
                    it.id == message.id
                }.subscribe {
                    sink.success(it)
                    disposable?.dispose()
                }
            }.timeout(Duration.ofSeconds(5))
        } else Mono.error(RuntimeException("发送消息失败"))
    }.log()
}

